package conf2.net;

import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.logging.Level;
import java.util.logging.Logger;

class Feeder
{

    private final ConcurrentLinkedQueue<Message> incoming,  outgoing;
    private boolean isRunning;
    private final ObjectInputStream in;
    private final ObjectOutputStream out;
    private final Logger logger = Logger.getLogger(Feeder.class.getName());
    private final IFeederErrorListener listener;
    private final Runnable sender,  receiver;

    Feeder(ObjectInputStream istream, ObjectOutputStream ostream, IFeederErrorListener listener)
    {
        incoming = new ConcurrentLinkedQueue<Message>();
        outgoing = new ConcurrentLinkedQueue<Message>();

        this.in = istream;
        this.out = ostream;
        this.listener = listener;
        isRunning = true;
        sender = new Runnable()
        {

            public void run()
            {
                try {
                    while (isRunning) {

                        while (!outgoing.isEmpty()) {
                            Message msg = outgoing.poll();
                            try {
                                out.writeObject(msg);
                                if (msg.getType() == Message.Type.PEER_IS_DISCONNECTING) {
                                    isRunning = false;
                                    break;
                                }
                            } catch (Exception e) {
                                if (isRunning) {
                                    error(e);
                                    isRunning = false;
                                }
                                break;
                            }
                        }
                        try {
                            Thread.sleep(200);
                        } catch (InterruptedException e) {
                            // do nothing
                        }
                    }
                } finally {
                    try {
                        out.close();
                    } catch (Exception e) {
                        logger.log(Level.WARNING, "Error while closing output stream.", e);
                    }
                }
            }
        };

        receiver = new Runnable()
        {

            public void run()
            {
                try {
                    while (isRunning) {
                        try {
                            Message m = (Message) in.readObject();
                            if (m.getType() == Message.Type.PEER_IS_DISCONNECTING) {
                                isRunning = false;
                                break;
                            } else {
                                incoming.add(m);

                            }
                        } catch (Exception e) {
                            if (isRunning) {
                                error(e);
                                isRunning = false;
                            }
                        }

                        try {
                            Thread.sleep(200);
                        } catch (InterruptedException e) {
                            // do nothing
                        }
                    }
                } finally {

                    try {
                        in.close();
                    } catch (Exception e) {
                        logger.log(Level.WARNING, "Error while closing input stream.", e);
                    }
                }
            }
        };

        Thread senderThread = new Thread(sender);
        Thread receiverThread = new Thread(receiver);
        senderThread.start();
        receiverThread.start();
    }

    private void error(Exception ex)
    {
        logger.log(Level.SEVERE, "Connection has been broken!", ex);
        listener.feederError();
        isRunning = false;
    }

    void terminate()
    {
        Message m = new Message(Message.Type.PEER_IS_DISCONNECTING, null);
        queueMessage(m);
        try {
            Thread.sleep(300);
        } catch (InterruptedException ie) {
        }
        isRunning = false;
    }

    void queueMessage(Message msg)
    {

        outgoing.add(msg);
    }

    Message getNextMessage()
    {
        Message m = null;
        m = incoming.peek();
        return m;
    }

    void removeFirstIncoming()
    {
        incoming.remove();
    }
}

